package cn.com.niit.socketio;

import com.corundumstudio.socketio.SocketIOClient;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
/**
 * 为每个连接的客户端在后台单独开启一个线程，此线程读取kafka数据，并使用socketio的SocketIOClient家境数据发送给客户端
 */
public class SocketThread extends Thread{
    //SocketIOClient用于服务器向客户端发送消息
    private SocketIOClient client;
    //flag用于标识SocketIOClient是否可用，初始值为true
    private boolean flag;

    public SocketThread(SocketIOClient client) {
        this.client = client;
        flag = true;
    }

    @Override
    public void run() {
        // 1.创建Kafka消费者配置
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "8.152.219.6:9092");
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "niit4444");
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 2.创建Kafka消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
        // 3. 订阅要消费的主题
        // 指定消费者从哪个topic中拉取数据
        kafkaConsumer.subscribe(Arrays.asList("term2"));
        // 4.使用一个while循环，不断从Kafka的topic中拉取消息
        while (flag) {
            // Kafka的消费者一次拉取一批的数据
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5));
            // 5.循环，获取value
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                String value = consumerRecord.value();
                System.out.println(value);
                //6、服务器向客户端发送消息
                if(client!=null && client.isChannelOpen()){
                    //client.sendEvent("msg",result);
                    client.sendEvent("termDate",value);
                }else{
                    flag = false;
                }
            }
        }
        //7、关闭
        kafkaConsumer.close();
        System.out.println("线程结束...");
    }
}
